package com.bw.gmall.realtime.app.dwd;

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import com.bw.gmall.realtime.utils.MysqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.hadoop.hbase.util.Sleeper;

public class DwdTradeCartAdd {
    public static void main(String[] args) throws Exception {

        //todo 1.初始化上下文
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //todo 2.设置并行度
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        //todo 3.使用ddl方式读取 topic_db 主题数据创建表
        tEnv.executeSql(MyKafkaUtil.getTopicDb("card_add"));


        Table cartAddTable = tEnv.sqlQuery("" +
                "select " +
                "    `data`['id'] id, " +
                "    `data`['user_id'] user_id, " +
                "    `data`['sku_id'] sku_id, " +
                "    `data`['cart_price'] cart_price, " +
                "    if(`type`='insert',`data`['sku_num'],cast(cast(`data`['sku_num'] as int) - cast(`old`['sku_num'] as int) as string)) sku_num, " +
                "    `data`['sku_name'] sku_name, " +
                "    `data`['is_checked'] is_checked, " +
                "    `data`['create_time'] create_time, " +
                "    `data`['operate_time'] operate_time, " +
                "    `data`['is_ordered'] is_ordered, " +
                "    `data`['order_time'] order_time, " +
                "    `data`['source_type'] source_type, " +
                "    `data`['source_id'] source_id, " +
                "    pt " +
                "from topic_db " +
                "where `database` = 'gmall' " +
                "and `table` = 'cart_info' " +
                "and (`type` = 'insert' " +
                "or (`type` = 'update'  " +
                "    and  " +
                "    `old`['sku_num'] is not null  " +
                "    and  " +
                "    cast(`data`['sku_num'] as int) > cast(`old`['sku_num'] as int)))");


        tEnv.createTemporaryView("cart_info_table",cartAddTable);

//        tEnv.toAppendStream(cartAddTable, Row.class).print(">>>>>");
//        tEnv.sqlQuery("select * from cart_info_table").execute().print();


        //todo 4.读取mysql中base_dic 表

        tEnv.executeSql(MysqlUtil.getBaseDicLookUpDDL());
//
//        //todo 5.关联2张表
//        /*
//        select
//
//        from cart_info_table ci
//        JOIN base_dic FOR SYSTEM_TIME AS OF ci.pt AS dic
//        on ci.source_type = dic.dic_code
//         */
        Table cartAddDicTable = tEnv.sqlQuery("select  " +
                "ci.id, " +
                "ci.user_id, " +
                "ci.sku_id, " +
                "ci.cart_price, " +
                "ci.sku_num, " +
                "ci.sku_name, " +
                "ci.is_checked, " +
                "ci.create_time, " +
                "ci.operate_time, " +
                "ci.is_ordered, " +
                "ci.order_time, " +
                "ci.source_type  source_type_id, " +
                "dic.dic_name  source_type_name, " +
                "ci.source_id " +
                "from cart_info_table ci " +
                "JOIN base_dic FOR SYSTEM_TIME AS OF ci.pt AS dic  " +
                "on ci.source_type = dic.dic_code ");

        tEnv.createTemporaryView("cart_add_dic_table",cartAddDicTable);
//
//        //todo 6.创建加购事实表
        tEnv.executeSql("create table dwd_cart_add( " +
                "id string, " +
                "user_id string, " +
                "sku_id string, " +
                "cart_price string, " +
                "sku_num string, " +
                "sku_name string, " +
                "is_checked string, " +
                "create_time string, " +
                "operate_time string, " +
                "is_ordered string, " +
                "order_time string, " +
                "source_type_id string, " +
                "source_type_name string, " +
                "source_id string " +
                ") " + MyKafkaUtil.getKafkaSinkDDL("dwd_trade_cart_add"));

        tEnv.executeSql("insert into dwd_cart_add select * from cart_add_dic_table");

        env.execute();
    }
}
